package me.riverslob.test.esper.plugin.esperio.rpc.output;

import com.espertech.esper.adapter.BaseSubscription;
import com.espertech.esper.client.EventBean;
import com.espertech.esper.event.EventBeanReader;
import com.espertech.esper.event.EventTypeSPI;
import com.espertech.esper.filter.FilterHandleCallback;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.*;

/**
 * <br/>Created on 13-10-12 下午5:20
 *
 * @author 郑治明（zm.zheng@zhuche.com）
 * @since 3.2.2
 */
public class EsperIORPCSubscription extends BaseSubscription {
    private static final Logger log = LoggerFactory.getLogger(EsperIORPCSubscription.class);

    private static final String OUTPUT_SERVICE_NAME = "esperRpcService";
    private String uuid = UUID.randomUUID().toString();
    private final String stream;
    private final List<String> projects;

    public EsperIORPCSubscription(String stream, List<String> projects) {
        this.stream = stream;
        this.projects = projects;
    }

    public void matchFound(EventBean theEvent, Collection<FilterHandleCallback> allStmtMatches) {
        /*
        EventTypeSPI spi = (EventTypeSPI) theEvent.getEventType();
        EventBeanReader reader = spi.getReader();
        Object[] props = reader.read(theEvent);
        String names[] = spi.getPropertyNames();
        Map<String, String> dataMap = formPairs(names, props, "stream", stream);
        */
        Map<String, Object> dataMap = new HashMap<String, Object>();
        dataMap.put("stream",stream);
        dataMap.put("data", theEvent);

        for (String project : projects) {
            try {
                if (log.isDebugEnabled()) {
                    log.debug("Output stream " + stream + " to project" + project);
                }
//                RemoteClientFactory.getInstance().asynExecuteToObject(project + OUTPUT_SERVICE_NAME, dataMap);
            } catch (Exception e) {
                log.error("ERROR Output stream " + stream + " to project" + project, e);
            }
        }

    }

    public boolean isSubSelect() {
        return false;
    }

    public String getStatementId() {
        return uuid;
    }

    private static Map<String, String> formPairs(String[] properties, Object[] values, String... additional) {
        Map<String, String> map = new LinkedHashMap<String, String>();
        for (int i = 0; i < additional.length; i += 2) {
            String key = additional[i];
            String val = i + 1 < additional.length ? additional[i + 1] : "";
            map.put(key, val);
        }
        for (int i = 0; i < properties.length; i++) {
            String key = properties[i];
            Object value = values[i];
            if (value == null) {
                map.put(key, "");
            } else {
                map.put(key, value.toString());
            }
        }
        return map;
    }
}
